package com.lmq.sparkConcept;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import scala.Tuple2;

import java.util.*;
import java.util.stream.Collectors;

/**
 * A simple spark app in java
 */
public class firstAppJavaversion {
    public static void main(String[] args) {
        JavaSparkContext sc = new JavaSparkContext("local[2]",
                "First spark app");
        // we take the row data in CSV format and convert it into
        //        a set of records fo the form (user, product, price)
        JavaRDD<String[]> data = sc.textFile("data/UserPurchaseHistory.csv")
                .map(s -> s.split(","));

        // lets count the number of the purchases
        long numPurchases = data.count();
        // lets count how many unique users made purchase
        long uniqueUsers = data.map(strings ->
                strings[0]).distinct().count();
        // lets sum up total revenue
        Double totalRevenue = data.map(strings ->
                Double.parseDouble(strings[2])).reduce((Double v1,
                                                        Double v2) -> v1 + v2);

        // lets find our most popular product
        List<Tuple2<String, Integer>> pairs = data.mapToPair(strings -> new Tuple2<String, Integer>(strings[1], 1)).reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        }).collect();

        Map<String, Integer> sortedData = new HashMap<>();
        Iterator it = pairs.iterator();

        while (it.hasNext()){
            Tuple2<String, Integer> o = (Tuple2<String, Integer>) it.next();
            sortedData.put(o._1,o._2);

        }
        List<String> sorted = sortedData.entrySet()
                .stream()
                .sorted(Comparator.comparing((Map.Entry<String, Integer>
                                                      entry) -> entry.getValue()).reversed())
                .map(Map.Entry::getKey)
                .collect(Collectors.toList());
        String mostPopular = sorted.get(0);
        Integer purchases = sortedData.get(mostPopular);
        System.out.println("Total purchases: " + numPurchases);
        System.out.println("Unique users: " + uniqueUsers);
        System.out.println("Total revenue: " + totalRevenue);
        System.out.println(String.format("Most popular product: %s with %d purchases", mostPopular, purchases));


    }

}
